package com.runa.monitor.platinterface.meter3_0.task;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.runa.monitor.platinterface.meter3_0.command.ReqControlValve;
import com.runa.monitor.platinterface.meter3_0.command.ReqSendBuffer;
import com.runa.monitor.platinterface.meter3_0.handle.AckMessageHandler;
import com.runa.monitor.platinterface.meter3_0.session.SessionManager;
import com.runa.monitor.platinterface.meter3_0.session.SessionProxy;
import com.runa.monitor.platinterface.meter3_0.session.WebSocketFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import javax.websocket.Session;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;

/**
 * @author diandian
 * @Date 2022/11/20 17:50
 */
@Slf4j
@Component
public class TaskManager {

    @Resource
    private AckMessageHandler ackMessageHandler;

    @Resource
    private WebSocketFactory webSocketFactory;
    @Resource
    private SessionManager sessionManager;

    protected Map<String, ConcurrentLinkedQueue<Task>> commandMap = new ConcurrentHashMap<>();

    protected CopyOnWriteArrayList<TaskProxy> currentCommandList = new CopyOnWriteArrayList<>();

    private Thread executeThread;

    public TaskManager() {
/*        ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
        //service.scheduleAtFixedRate(this::consumption, 0, 1, TimeUnit.MICROSECONDS);
        service.execute(this::consumption);*/

        Runnable runnable = this::consumption;
        this.executeThread = new Thread(runnable);
        executeThread.start();
    }

    public static void main(String[] args) {
        long convert = TimeUnit.SECONDS.toNanos(10);
        System.out.println(convert);
    }

    private void consumption() {
        log.info("琅卡博3.0档案下发---consumption 10"+ JSON.toJSONString(currentCommandList));
        long convert = TimeUnit.SECONDS.toNanos(10);
//        log.info("琅卡博3.0档案下发---consumption 11"+ JSON.toJSONString(convert));
        for (; ; ) {
            for (TaskProxy taskProxy : currentCommandList) {
                if (taskProxy.getTaskState()) {
                    //已经发送了数据还未收到响应
                    continue;
                }

                boolean sendSuccess = this.handle(taskProxy);
                if (sendSuccess) {
                    taskProxy.setTaskState(true);
                }
            }
            //锁定10s检查一次，如果有唤醒就提前检查，按道理是不需要轮询检查的
            LockSupport.parkNanos(convert);
        }
    }

    //入队
    public void offer(String netEquNo, Task task) {
        ConcurrentLinkedQueue<Task> ctrlValveTasks = commandMap.computeIfAbsent(netEquNo, t -> new ConcurrentLinkedQueue<>());
        ctrlValveTasks.offer(task);
        SessionProxy sessionProxy = this.getSessionProxy(netEquNo);
        if (sessionProxy == null) {
            log.error("琅卡博3.0无法创建会话");
            //无法创建会话
            return;
        }
        SessionProxy.SessionState sessionState = sessionProxy.getSessionState();
        log.info("琅卡博3.0档案下发---7"+ JSON.toJSONString(sessionState));
        //如果当前会话是空闲状态，直接发送数据到待发区
        if (SessionProxy.SessionState.free.equals(sessionState)) {
            this.moveToCurrent(netEquNo);
        }
    }

    public void removeCurrent(TaskProxy taskProxy, String netEquNo) {
        //响应结束移除当前的数据
        currentCommandList.remove(taskProxy);
        //因为移除当前数据，说明当前会话空闲了，需要有新的任务进去
        this.moveToCurrent(netEquNo);
    }

    private void moveToCurrent(String netEquNo) {
        ConcurrentLinkedQueue<Task> ctrlValveTasks = commandMap.get(netEquNo);
        log.info("琅卡博3.0档案下发---8"+ JSON.toJSONString(ctrlValveTasks));
        if (ctrlValveTasks == null) {
            return;
        }
        Task take = ctrlValveTasks.poll();
        if (take == null) {
            return;
        }
        TaskProxy taskProxy = new TaskProxy();
        taskProxy.setTaskState(false);
        taskProxy.setTask(take);
        currentCommandList.add(taskProxy);
        log.info("琅卡博3.0档案下发---10"+ JSON.toJSONString(currentCommandList)+"---currentCommandList---"+JSON.toJSONString(currentCommandList));
        //有新的数据进入到待发送区，触发发送线程工作
        LockSupport.unpark(executeThread);
    }


    private boolean handle(TaskProxy taskProxy) {
        Task task = taskProxy.getTask();
        SessionProxy sessionProxy = this.getSessionProxy(task.getNetEquNo());
        if (sessionProxy == null) {
            return false;
        }
        boolean res;
        Session session = sessionProxy.getSession();
        String message = null;
        if (task instanceof CtrlValveTask) {
            message = ctrlValveTaskMsg((CtrlValveTask) task);
        }
        if (task instanceof SendBufferTask) {
            message = sendBufferTaskMsg((SendBufferTask) task);
        }
        try {
            session.getBasicRemote().sendText(message);
            sessionProxy.setResponseBusinessHandler(ackMessageHandler);
            sessionProxy.setTaskProxy(taskProxy);
            sessionProxy.setTaskFunction(task.getFunction());
            //标记当前会话已经发送了一条数据，需要等待响应
            log.info("WEBSocket消息发送---"+ message);
            sessionManager.notifySendMsg(task.getNetEquNo(), message);
            res = true;
        } catch (IOException e) {
            res = false;
            
        }
        return res;
    }

    private String ctrlValveTaskMsg(CtrlValveTask task) {
        ReqControlValve.Data data = new ReqControlValve.Data();
        List<ReqControlValve.ValvesInfo> valves = new ArrayList<>();
        data.setValves(valves);
        ReqControlValve command = new ReqControlValve(task.getNetEquNo());
        command.setData(data);

        for (String equNo : task.getEquNo()) {
            ReqControlValve.ValvesInfo valvesInfo = new ReqControlValve.ValvesInfo(equNo, String.valueOf(task.getOpening()));
            valvesInfo.setIsLock(task.getIfLock());
            valves.add(valvesInfo);
        }
        return JSON.toJSONString(command);
    }

    private String sendBufferTaskMsg(SendBufferTask task) {
        ReqSendBuffer.Data data = new ReqSendBuffer.Data();
        data.setBuffer(task.getHexBuffer());
        ReqSendBuffer command = new ReqSendBuffer(task.getNetEquNo());
        command.setData(data);
        return JSON.toJSONString(command);
    }

    private SessionProxy getSessionProxy(String netEquNo) {
        //type = 1 表示是构建计量的响应处理类
        SessionProxy sessionProxy = webSocketFactory.createSession(netEquNo, 1);
        if (sessionProxy == null) {
            return null;
        }
        //判断当前session的状态，空闲状态才可以继续发送数据
        SessionProxy.SessionState sessionState = sessionProxy.getSessionState();
        if (!sessionState.equals(SessionProxy.SessionState.free)) {
            log.error("WEBSocket SessionProxy当前正忙---");
            return null;
        }
        return sessionProxy;
    }


//    public static void main(String[] args) {
//        ReqControlValve.Data data = new ReqControlValve.Data();
//        List<ReqControlValve.ValvesInfo> valves = new ArrayList<>();
//        data.setValves(valves);
//        ReqControlValve command = new ReqControlValve("netEquNo");
//        command.setData(data);
//
//        ReqControlValve.ValvesInfo valvesInfo = new ReqControlValve.ValvesInfo("netEquNo", String.valueOf("999"));
//        valves.add(valvesInfo);
//
//        System.out.println(JSONObject.toJSONString(command));
//    }
}
